package com.shujia.streaming

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, Statement}
import scala.collection.mutable

object Demo04BuKong {
  def main(args: Array[String]): Unit = {
    /**
     * 缉查布控:
     * 在MySQL中有一份黑名单:
     * create table mdn_blacklist(
     * mdn varchar(255) comment '人员mdn'
     * );
     * 插入数据：
     * insert into mdn_blacklist values
     * ('47BE1E866CFC071DB19D5E1C056BE28AE24C16E7'),
     * ('6A1F2758502F63D20E47D9359E8628255E531363'),
     * ('CA66090AFD6273647237ECDB290474C83803F808'),
     * ('5B6559EC824C9DE12A28C2C617FEC2188D96D4FC'),
     * ('B90249CDA092CE9C939424479A0C0841D2C01660'),
     * ('B2B66493E78763D764143792D2237FFD8211049A'),
     * ('AF3ADA85C95EA54D57ED0515CEBBD8958E791F73'),
     * ('A8B7AEB1514F191DF5212B17E5D97D0EC64E83DA');
     *
     * 使用nc -lk 8888 模拟消息队列 手动的“采集”数据
     *
     * 使用SparkStreaming实时处理数据 一旦发现数据中有黑名单中提供的人员
     * 即将数据过滤出来并写入MySQL
     *
     * 为什么不用本地的List而用MySQL去保存黑名单信息？
     * 需要动态的增加或删除黑名单中的人员，如果直接使用本地的List则需要重启应用
     */

    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo04BuKong")
      .master("local[2]")
      .getOrCreate()

    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    val lineDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    /**
     * 从MySQL获取黑名的信息
     */

    // 构建一个集合Set用于保存从MySQL获取到的黑名单 并用于过滤
    //    val mdnSet: mutable.HashSet[String] = mutable.HashSet()
    //
    //    val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bukong", "root", "123456")
    //    val stat: Statement = conn.createStatement()
    //    val rs: ResultSet = stat.executeQuery("select mdn from mdn_blacklist")
    //    while (rs.next()) {
    //      val mdn: String = rs.getString("mdn")
    //      mdnSet.add(mdn)
    //    }
    //    stat.close()
    //    conn.close()

    /**
     * 一个Batch会封装成一个RDD进行处理
     * 问题：
     * 1、需要实现动态的增加删除黑名单中的人员
     * 2、性能问题 连接在哪里创建？创建多少次？
     *
     */
    //    lineDS
    //      .filter(line => {
    //
    //        /**
    //         * 直接放在filter算子内部
    //         * 相当于每个Batch中的每条记录都会创建一次连接 效率很低
    //         *
    //         */
    //        val mdnSet: mutable.HashSet[String] = mutable.HashSet()
    //
    //        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bukong", "root", "123456")
    //        val stat: Statement = conn.createStatement()
    //        val rs: ResultSet = stat.executeQuery("select mdn from mdn_blacklist")
    //        while (rs.next()) {
    //          val mdn: String = rs.getString("mdn")
    //          mdnSet.add(mdn)
    //        }
    //        stat.close()
    //        conn.close()
    //
    //        // 从数据中提取mdn 与 黑名单进行比较
    //        val mdn: String = line.split(",")(0)
    //        mdnSet.contains(mdn)
    //      })
    //      .print()
    val filterDS: DStream[String] = lineDS
      // 每个Batch会执行一次 可以将Batch变成RDD进行处理
      // transform的作用实际上是让我们能够使用RDD的API去处理数据
      .transform(rdd => {
        // 使用RDD的API进行处理
        // 每个Batch 即 每5s会建立一次连接
        val mdnSet: mutable.HashSet[String] = mutable.HashSet()
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bukong", "root", "123456")
        val stat: Statement = conn.createStatement()
        val rs: ResultSet = stat.executeQuery("select mdn from mdn_blacklist")
        while (rs.next()) {
          val mdn: String = rs.getString("mdn")
          mdnSet.add(mdn)
        }
        stat.close()
        conn.close()
        // 算子内部使用了算子外部的集合 可以进行广播
        val mdnSetBro: Broadcast[mutable.HashSet[String]] = rdd.sparkContext.broadcast(mdnSet)

        rdd.mapPartitions(iter => {
          // 将建立连接放在每个Batch内部的RDD然后进行mapPartitions
          // 相当于每个Batch中的RDD的每个分区会建立一次连接

          iter.filter(line => {
            // 从数据中提取mdn 与 黑名单进行比较
            val mdn: String = line.split(",")(0)
            mdnSetBro.value.contains(mdn)
          })
        })
      })
    // 将结果写入MySQL
    filterDS

      /**
       * foreachRDD vs transform
       * 都可以将每个Batch即DStream变成一个个的RDD进行处理
       * 如果要写入外部系统 ===> foreachRDD
       * 如果相对数据进行转换 ===> transform
       */
      .foreachRDD(rdd => {
        // 这里的代码还是会在Driver端执行 所以不能在此建立连接
        rdd.foreachPartition(iter => {
          val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bukong", "root", "123456")
          val preSt: PreparedStatement = conn.prepareStatement("insert into blacklist_info values (?,?,?,?)")

          iter.foreach(line => {
            val splits: Array[String] = line.split(",")
            val mdn: String = splits(0)
            val start_date: String = splits(1)
            val longitude: String = splits(4)
            val latitude: String = splits(5)
            preSt.setString(1, mdn)
            preSt.setString(2, start_date)
            preSt.setString(3, longitude)
            preSt.setString(4, latitude)
            preSt.addBatch() // 将RDD中的每个分区作为一个批次 使用批量插入
          })
          preSt.executeBatch()
          preSt.close()
          conn.close()
        })

      })

    // 启动
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}
